package com.atguigu.flink.state;

import com.atguigu.flink.pojo.HotItem;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by 黄凯 on 2023/6/22 0022 17:30
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 需求： 每隔5min输出最近1h内点击量(pv)最多的前3个商品
 * *
 * *  数据:    543462,1715,1464116,pv,1511658000
 * *          userId,商品id,商品类别id,行为类型，ts
 * *
 * *   输入:  543462,1715,1464116,pv,1511658000
 * *          粒度： 一个用户点击一个商品的一次是一行
 * *
 * *
 * *   推理计算过程:   聚合，keyBy  商品id
 * *          第一次聚合:  统计最近1h(窗口)内，各个商品的点击总次数
 * *              size: 范围，1h
 * *              slide: 计算时机，5min
 * *                  滑动的时间窗口。
 * *
 * *              输入:  543462,1715,1464116,pv,1511658000
 * *
 * *              输出:
 * *                    [8:00,9:00)：
 * *                          A---120
 * *                    [8:05,9:05)
 * *                          A---150
 * *                    [8:00,9:00)：
 * *                          B---130
 * *                    [8:00,9:00)：
 * *                          C---132
 * *                    [8:00,9:00)：
 * *                          D---131
 * *
 * *          第二次聚合:  将每个时间段窗口中各个商品的点击量，排序再取前3
 * *                  用不用开窗? 不用
 * *                  需要keyBy，按照窗口的统计的时间范围keyBy
 * *
 * *                  等同一个窗口的所有数据全部到达后，再一次性计算。
 * *                  如何知道当前要计算的数据已经全部到达，可以触发运算?
 * *                      使用定时器，将窗口的endTime作为触发时间，只要下游的时间到了endTime证明上游endTime之前的所有数据都已经到达了下游，可以进行运算。
 * *
 * *
 * *
 * *    输出:
 * *              [8:00,9:00):
 * *                      A--120
 * *                      B--119
 * *                      C--118
 * *              [8:05,9:05):
 * *                       E--120
 * *                       B--119
 * *                       C--118
 */
public class Flink12_TopN {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                new TextLineInputFormat(StandardCharsets.UTF_8.name())
                ,
                new Path("input/UserBehavior.csv")
        ).build();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp() * 1000);

        //读数据，封装bean，过滤pv，生成水印
        SingleOutputStreamOperator<UserBehavior> ds = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "source")
                .map(line -> {

                    String[] words = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(words[0]),
                            Long.valueOf(words[1]),
                            Integer.valueOf(words[2]),
                            words[3],
                            Long.valueOf(words[4])
                    );

                }).filter(
                        bean -> "pv".equals(bean.getBehavior())
                ).assignTimestampsAndWatermarks(watermarkStrategy);

        /*
            2.开窗，统计每种商品的点击次数

         */
        SingleOutputStreamOperator<HotItem> ds1 = ds
                .keyBy(
                        UserBehavior::getItemId
                ).window(
                        SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))
                ).aggregate(new AggregateFunction<UserBehavior, Long, HotItem>() {
                                @Override
                                public Long createAccumulator() {
                                    return 0L;
                                }

                                @Override
                                public Long add(UserBehavior value, Long accumulator) {
                                    return accumulator + 1;
                                }

                                @Override
                                public HotItem getResult(Long accumulator) {
                                    return new HotItem(null, null, null, accumulator);
                                }

                                @Override
                                public Long merge(Long a, Long b) {
                                    return null;
                                }
                            },
                        new ProcessWindowFunction<HotItem, HotItem, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                ProcessWindowFunction<HotItem, HotItem, Long, TimeWindow>.Context context,
                                                Iterable<HotItem> elements,
                                                Collector<HotItem> out) throws Exception {

                                HotItem hotItem = elements.iterator().next();

                                hotItem.setItemId(key);
                                hotItem.setStart(context.window().getStart());
                                hotItem.setEnd(context.window().getEnd());

                                out.collect(hotItem);

                            }
                        }
                );

        //3. 在下游按照窗口的时间范围分组，top3统计。使用定时器触发运算
        ds1.keyBy(
                HotItem::getStart
        ).process(

                new KeyedProcessFunction<Long, HotItem, String>() {

                    private ValueState<Boolean> flag;
                    private ListState<HotItem> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        listState = getRuntimeContext().getListState(new ListStateDescriptor<>("hot3", HotItem.class));

                        flag = getRuntimeContext().getState(new ValueStateDescriptor<>("flag", Boolean.class));

                    }

                    /**
                     * 进行top3计算
                     */
                    @Override
                    public void onTimer(long timestamp,
                                        KeyedProcessFunction<Long, HotItem, String>.OnTimerContext ctx,
                                        Collector<String> out) throws Exception {

                        List<HotItem> top3 = StreamSupport.stream(listState.get().spliterator(), true)
                                .sorted((o1, o2) -> -o1.getCount().compareTo(o2.getCount()))
                                .limit(3)
                                .collect(Collectors.toList());

                        //整理数据的格式
                        String resultStr = top3.stream()
                                .map(item -> item.getItemId() + ":" + item.getCount())
                                .collect(Collectors.joining(","));

                        String timeStr = MyUtil.parseTimeWindow(new TimeWindow(top3.get(0).getStart(), top3.get(0).getEnd()));

                        out.collect(timeStr + ": top3  : " + resultStr);

                    }

                    /**
                     * 每来一条数据，先存起来，等定时器到点了，再触发top3
                     */
                    @Override
                    public void processElement(HotItem hotItem,
                                               KeyedProcessFunction<Long, HotItem, String>.Context context,
                                               Collector<String> collector) throws Exception {

                        listState.add(hotItem);

                        //在当前组中第一条数据来的时候，注册定时器
                        if (flag.value() == null) {

                            context.timerService().registerEventTimeTimer(hotItem.getEnd());

                            flag.update(false);

                        }

                    }
                }


        ).print();

        env.execute();


    }

}
